package com.gome.boot.config.mq.rocketmq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.*;
import com.alibaba.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.gome.rocketmq.client.extension.impl.MessageProducerImpl;
import com.gome.utils.DiamondUtil;
import com.gome.utils.RunTimeUtil;
import com.taobao.diamond.manager.ManagerListenerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 消息生产者扩展.
 *
 * @author baoxiufeng
 */
public class MessageProducerExtImpl extends MessageProducerImpl {
    private static final Logger logger = LoggerFactory.getLogger(MessageProducerExtImpl.class);
    private SerializeType serializeType;
    private int maxMessageSize;
    private String topic;
    private String tags;
    private DefaultMQProducer producer;
    private String dataId;
    private String group;
    private int level = 0;
    private boolean orderly;
    private MessageQueueSelector selector;

    public MessageProducerExtImpl() {
        this(false);
    }

    public MessageProducerExtImpl(boolean orderly) {
        this.serializeType = SerializeType.FASTJSON;
        this.maxMessageSize = 4194304;
        this.orderly = orderly;
        if (this.orderly) {
            this.selector = new SelectMessageQueueByHash();
        }
    }

    @Override
    public void destroy() throws Exception {
        this.producer.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.registDiamondListener();
    }

    /**
     * 注册Diamond监听器.
     */
    private void registDiamondListener() {
        DiamondUtil.registDiamondListener(new ManagerListenerAdapter() {
            @Override
            public void receiveConfigInfo(String configInfo) {
                MessageProducerExtImpl.this.restartProducer(configInfo);
            }
        }, this.group, this.dataId);
    }

    @Override
    public synchronized void restartProducer(String configInfo) {
        DefaultMQProducer producerNew = null;

        try {
            producerNew = new DefaultMQProducer(this.topic);
            producerNew.setNamesrvAddr(configInfo);
            producerNew.setMaxMessageSize(this.maxMessageSize);
            producerNew.setInstanceName(RunTimeUtil.getRocketMqUniqeInstanceName());
            producerNew.start();
        } catch (MQClientException var4) {
            logger.error("rocketmq restartProducer error ", var4);
            return;
        }

        DefaultMQProducer oldInsetance = this.producer;
        this.producer = producerNew;
        if (oldInsetance != null) {
            oldInsetance.shutdown();
        }
    }

    private SendResult send(Message msg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        SendResult sendResult = null;
        if (orderly) {
            sendResult = this.producer.send(msg, selector, msg.getKeys());
        } else {
            sendResult = this.producer.send(msg);
        }
        return sendResult;
    }

    private void send(Message msg, SendCallback sendCallback) throws RemotingException, MQClientException, InterruptedException {
        if (orderly) {
            this.producer.send(msg, selector, msg.getKeys(), sendCallback);
        } else {
            this.producer.send(msg, sendCallback);
        }
    }

    @Override
    public SendResult send(String key, Object messageObject) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        Message msg = new Message(this.topic, this.tags, key, this.serializeType.serialize(messageObject));
        msg.setDelayTimeLevel(level);
        SendResult sendResult = null;
        sendResult = send(msg);
        logger.info("rocketmq send success,key=" + key + ";sendResult:" + sendResult);
        return sendResult;
    }

    @Override
    public SendResult send(String key, byte[] message) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        Message msg = new Message(this.topic, this.tags, key, message);
        msg.setDelayTimeLevel(level);
        SendResult sendResult = null;
        sendResult = send(msg);
        logger.info("rocketmq send success,key=" + key + ";sendResult:" + sendResult);
        return sendResult;
    }

    @Override
    public void send(String key, Object messageObject, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
        Message msg = new Message(this.topic, this.tags, key, this.serializeType.serialize(messageObject));
        msg.setDelayTimeLevel(level);
        send(msg, sendCallback);
        logger.info("rocketmq send key=" + key);
    }

    @Override
    public int getMaxMessageSize() {
        return this.maxMessageSize;
    }

    @Override
    public void setMaxMessageSize(int maxMessageSize) {
        this.maxMessageSize = maxMessageSize;
    }

    @Override
    public void setTopic(String topic) {
        this.topic = topic;
    }

    @Override
    public void setTags(String tags) {
        this.tags = tags;
    }

    @Override
    public void setDataId(String dataId) {
        this.dataId = dataId;
    }

    @Override
    public void setGroup(String group) {
        this.group = group;
    }

    public void setLevel(int level) {
        this.level = level;
    }

    public void setSerializeType(SerializeType serializeType) {
        this.serializeType = serializeType;
    }

    public void setOrderly(boolean orderly) {
        this.orderly = orderly;
    }

    @Override
    public MQProducer getMQProducer() {
        return this.producer;
    }
}
